package com.markhsiu.minimq.zookeeper;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.TimeUnit;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooKeeper.States;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ZkClient {
	private static Logger logger = LoggerFactory.getLogger(ZkClient.class);
	private static final int VERSION = -1;
    // 会话超时时间，设置为与系统默认时间一致
    private static final int SESSION_TIMEOUT = 30 * 1000;

    // 创建 ZooKeeper 实例
    private ZooKeeper zk;

    // 创建 Watcher 实例
    private Watcher wh = new Watcher() {
        /**
         * Watched事件
         */
        public void process(WatchedEvent event) {
        	logger.info("WatchedEvent >>> {}" , event.toString());
        } 
    };
    
    private static ZkClient INSTANCE = new ZkClient();
    private ZkClient(){};
    public static ZkClient instance(){
    	try {
			INSTANCE.init();
		} catch (IOException e) {
			e.printStackTrace();
		}
    	return INSTANCE;
    }

    // 初始化 ZooKeeper 实例
    public void init() throws IOException {
        // 连接到ZK服务，多个可以用逗号分割写
        zk = new ZooKeeper("127.0.0.1:2181", ZkClient.SESSION_TIMEOUT, this.wh);
        //在连接的时候判断连接状态，如果未连接成功，程序自动使用其他连接去请求连接
        if(!zk.getState().equals(States.CONNECTED)){
            while(true){
                if(zk.getState().equals(States.CONNECTED)){
                	logger.info("zookeeper连接成功。。。");
                    break;
                }
                try {
                	logger.info("等待zookeeper连接中。。。");
                    TimeUnit.SECONDS.sleep(5);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    
    public void createAndCheck(String path,boolean persistent) throws KeeperException, InterruptedException{
    	if(zk.exists(path, false) == null){
    		zk.create(path, null, Ids.OPEN_ACL_UNSAFE, checkMode(persistent));
    	}
    }
    
    public void createAndCheck(String path,byte[] data, boolean persistent) throws KeeperException, InterruptedException{
    	if(zk.exists(path, false) == null){
    		zk.create(path, data, Ids.OPEN_ACL_UNSAFE, checkMode(persistent));
    	}
    }
    
    public void create(String path,boolean persistent) throws KeeperException, InterruptedException{   	
    	zk.create(path, null, Ids.OPEN_ACL_UNSAFE, checkMode(persistent));
    }
    
    public void create(String path,byte[] data, boolean persistent) throws KeeperException, InterruptedException{
    	zk.create(path, data, Ids.OPEN_ACL_UNSAFE, checkMode(persistent));
    }
    
    private CreateMode checkMode(boolean persistent){
    	if(persistent){
    		return CreateMode.PERSISTENT;
    	}
    	
    	return CreateMode.EPHEMERAL;
    }
    public void delete(String path) throws InterruptedException, KeeperException{
    	zk.delete(path, VERSION);
    }
    
    public String getData(String path) throws InterruptedException, KeeperException{
    	byte[] data = zk.getData(path, false, null);
    	return data == null ? null : new String(data);
    }
    
    public void setData(String path,String data ) throws InterruptedException, KeeperException{
    	zk.setData(path,data == null ? null : data.getBytes(), VERSION);
    }
    
    public boolean exists(String path) throws InterruptedException, KeeperException{
    	return zk.exists(path, false) != null;
    }
   
    public List<String> getChildren(String path)
            throws KeeperException, InterruptedException {
        return zk.getChildren(path, false);
    }
    
    public void ZKClose() throws InterruptedException {
        zk.close();
    }
    
    
}
